package com.bizzan.bitrade.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.google.common.collect.Lists;
import com.rzcode.coinex.kernel.common.enums.BooleanEnum;
import com.rzcode.coinex.kernel.dao.entity.Coin;
import com.rzcode.coinex.kernel.dao.entity.Member;
import com.rzcode.coinex.kernel.dao.entity.MemberWallet;
import com.rzcode.coinex.kernel.dao.mapper.MemberWalletMapper;
import com.rzcode.coinex.kernel.service.CoinService;
import com.rzcode.coinex.kernel.service.MemberService;
import com.rzcode.coinex.kernel.service.MemberWalletService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
public class CoinConsumer {

    private final CoinService coinService;

    private final MemberWalletMapper memberWalletMapper;
    private final RestTemplate restTemplate;
    private final MemberService memberService;

    /**
     * 添加新币种，为用户增加钱包
     *
     * @param content 包含用户和币种信息的JSON字符串。
     */
    @KafkaListener(topics = {"coin-start"})
    public void handle(String content) {
        log.info("Handling coin data: {}", content);
        if (StringUtils.isEmpty(content)) {
            return;
        }
        JSONObject json = JSON.parseObject(content);
        if (json == null) {
            return;
        }

        String coinName = json.getString("name");
        Coin coin = coinService.getOne(coinName);
        if (coin != null && coin.getEnableRpc().equals(BooleanEnum.IS_TRUE)) {
            List<Member> members = memberService.listMembers();
            int batchSize = 5000; // 调整批量大小
            List<List<Member>> memberBatches = Lists.partition(members, batchSize);
            memberBatches.forEach(batch -> fetchAndSaveWallets(coin, batch)); // 传入 coin 和成员批次        }
        }
    }

    @Retryable(value = {Exception.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000))
    private void fetchAndSaveWallets(Coin coin, List<Member> members) {
        List<String> list1 = members.stream().map(member -> "U" + member.getId()).collect(Collectors.toList());
        String serviceName = "SERVICE-RPC-" + coin.getUnit();
        String url = "http://" + serviceName + "/rpc/address/batch";

        ResponseEntity<List> result = restTemplate.postForEntity(url, list1, List.class);
        if (result.getStatusCode().value() == 200) {
            List<Map> mapList = result.getBody();
            List<MemberWallet> wallets = mapList.stream()
                    .map(map -> {
                        MemberWallet wallet = new MemberWallet();
                        wallet.setCoin(coin);
                        wallet.setMemberId((Long) map.get("uid"));
                        wallet.setBalance(BigDecimal.ZERO);
                        wallet.setFrozenBalance(BigDecimal.ZERO);
                        wallet.setAddress((String) map.get("address"));
                        return wallet;
                    })
                    .collect(Collectors.toList());
            memberWalletMapper.saveBatch(wallets);
        } else {
            throw new RuntimeException("Remote service call failed with status: " + result.getStatusCodeValue());
        }
    }
}
